/**
 * 
 */
package org.tcp.client.pool.imp;

import java.io.IOException;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.tcp.client.exception.ExceptionEnum;
import org.tcp.client.exception.PoolException;
import org.tcp.client.pool.ITcpConnection;
import org.tcp.client.pool.Pool;
import org.tcp.client.pool.PoolConfig;
import org.tcp.client.pool.PoolConfig.PoolConfigEnum;
import org.tcp.client.pool.PoolHeat;
import org.tcp.client.pool.TcpConnection;

/**
 * @author kucheng.liao@autonavi.com
 * @Date Sep 19, 2014
 */
public class PoolImpl implements Pool {

	protected static Log log = LogFactory.getLog(PoolImpl.class);

	private int max_size;

	private int min_size;

	private AtomicInteger total;

	private AtomicInteger size;

	private volatile boolean isInit = true;

	private Map<PoolConfigEnum, String> configMap = new HashMap<PoolConfigEnum, String>();

	protected BlockingQueue<ITcpConnection> queue;

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.tcp.client.pool.Pool#poll()
	 */
	public ITcpConnection poll() throws PoolException {

		if (log.isDebugEnabled()) {
			log.debug("poll: the connections nubmer of pool , total  : " + total + "  size : " + size);
		}
		ITcpConnection connection = null;
		try {
			while (true) {
				if (size.get() == 0 && total.get() < max_size) {

					if (log.isDebugEnabled()) {
						log.debug(" the pool is empty and the total less than maxSize, so it will create a new connection");
					}
					connection = createConnection();
					return connection;
				} else {
					connection = queue.poll(15, TimeUnit.SECONDS);

					if (connection != null) {
						size.addAndGet(-1);
						return connection;
					} else {
						if (log.isDebugEnabled()) {
							log.debug("can't get a connection from pool after 15 seconds, it will throw a timeout exception");
						}
						throw new PoolException(ExceptionEnum.TIME_OUT);
					}
				}
			}
		} catch (InterruptedException e) {
			throw new PoolException(e);
		}

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.tcp.client.pool.Pool#push(java.lang.Object)
	 */
	public void push(ITcpConnection connection) throws PoolException {

		if (log.isDebugEnabled()) {
			log.debug("push:the connections nubmer of pool , total  : " + total + "  size : " + size);
		}

		if (null != queue) {
			queue.offer(connection);
			size.addAndGet(1);
		} else {

			if (log.isDebugEnabled()) {
				log.debug("the queue is null, connection will be destory");
			}

			try {
				connection.destory();
			} catch (IOException e) {
				throw new PoolException(e);
			}
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.tcp.client.pool.Pool#destory()
	 */
	public void destory() {
		if (log.isInfoEnabled()) {
			log.info("destory the pool");
		}

		if (null != queue) {
			synchronized (queue) {
				Queue<ITcpConnection> tmp = queue;
				queue = null;
				if (tmp.size() > 0) {
					while (tmp.size() > 0) {
						ITcpConnection connection = tmp.poll();
						try {
							connection.destory();
						} catch (IOException e) {
							log.error("", e);
						}
					}
				}
			}
		}

	}

	/**
	 * 创建一个连接
	 * 
	 * @return
	 * @throws UnknownHostException
	 * @throws IOException
	 */
	protected ITcpConnection createConnection() {
		if (log.isInfoEnabled()) {
			log.info(" create a new connections");
		}
		try {
			String ip = configMap.get(PoolConfigEnum.IP);
			int port = Integer.parseInt(configMap.get(PoolConfigEnum.PORT));
			ITcpConnection connection = new TcpConnection(ip, port, this);
			total.addAndGet(1);
			return connection;
		} catch (Exception e) {
			throw new PoolException(e);
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.tcp.client.pool.Pool#init(org.tcp.client.pool.PoolConfig)
	 */
	public boolean init(PoolConfig config) throws UnknownHostException, IOException {
		if (null == config) {
			if (log.isWarnEnabled()) {
				log.warn(" PoolConfig is null,the init is failed, please check the parameter");
			}
			return false;
		}
		if (log.isInfoEnabled()) {
			log.info("init the pool with parametes : " + config.getConfig());
		}
		if (isInit && null != config && config.validate()) {

			synchronized (this) {
				if (isInit) {
					configMap.putAll(config.getConfig());
					max_size = Integer.parseInt(configMap.get(PoolConfigEnum.POOL_MAX));
					min_size = Integer.parseInt(configMap.get(PoolConfigEnum.POOL_MIN));
					size = new AtomicInteger(min_size);
					total = new AtomicInteger(min_size);
					BlockingQueue<ITcpConnection> queueTMP = new LinkedBlockingQueue<ITcpConnection>(max_size);
					String ip = configMap.get(PoolConfigEnum.IP);
					int port = Integer.parseInt(configMap.get(PoolConfigEnum.PORT));
					for (int i = 0; i < min_size; i++) {
						ITcpConnection connection = new TcpConnection(ip, port, this);
						queueTMP.add(connection);
					}
					queue = queueTMP;
					PoolHeat heat = new PoolHeatImpl(config, queueTMP);
					heat.check();
					isInit = false;
				}
			}
		}
		return isInit;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.tcp.client.pool.Pool#getCurrentSize()
	 */
	public int getTotalSize() {
		return total.get();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.tcp.client.pool.Pool#getMaxSize()
	 */
	public int getMaxSize() {
		return max_size;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.tcp.client.pool.Pool#getMinSize()
	 */
	public int getMinSize() {
		return min_size;
	}

	public String toString() {
		return "total : " + total + " size ： " + size;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.tcp.client.pool.Pool#getSize()
	 */
	public int getSize() {
		return size.get();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.tcp.client.pool.Pool#addSize(int)
	 */
	public void addSize(int i) {
		size.addAndGet(i);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.tcp.client.pool.Pool#decrementTotal()
	 */
	public void decrementTotal() {
		total.addAndGet(-1);
	}
}
